//
// Created by benxb on 2021/11/16.
//

#include <common/util/file_util.h>
#include "storage/disk/data_chunk_storage.h"
#include "common/exception.h"
#include "common/logger.h"
#include "common/config.h"
#include <cstring>

namespace benxdb {


DataChunkStorage::DataChunkStorage(const std::string &table_name) {
    data_file_ = GlobalConfig::DataFilePath(table_name);
    db_io_.open(data_file_, std::ios::binary | std::ios::in | std::ios::out);
    // directory or file does not exist
    if (!db_io_.is_open()) {
        db_io_.clear();
        // create a new file
        db_io_.open(data_file_, std::ios::binary | std::ios::trunc | std::ios::out);
        db_io_.close();
        // reopen with original mode
        db_io_.open(data_file_, std::ios::binary | std::ios::in | std::ios::out);
        if (!db_io_.is_open()) {
            throw Exception("can't open db file");
        }
    }
    file_size_ = FileUtil::GetFileSize(data_file_);
}

void DataChunkStorage::ShutDown() {
    db_io_.close();
}

void DataChunkStorage::AppendChunkData(const char *chunk_data, int64_t chunk_size) {
    // set write cursor to offset
    num_writes_ += 1;
    db_io_.seekp(0, std::ios_base::end);
    db_io_.write(chunk_data, chunk_size);
    // check for I/O error
    if (db_io_.bad()) {
        LOG_DEBUG("I/O error while writing");
        return;
    }
    // needs to flush to keep disk file in sync
    db_io_.flush();
    file_size_ += chunk_size;
}

void DataChunkStorage::ReadChunkData(int64_t chunk_offset, int64_t chunk_size, std::string& chunk_data) {
    // check if read beyond file length;
    if (chunk_offset > file_size_) {
        LOG_DEBUG("I/O error reading past end of file");
    } else {
        // set read cursor to offset
        chunk_data.resize(chunk_size);
        db_io_.seekp(chunk_offset);
        db_io_.read(const_cast<char *>(chunk_data.data()), chunk_size);
        if (db_io_.bad()) {
            LOG_DEBUG("I/O error while reading");
            return;
        }
        // if file ends before reading PAGE_SIZE
        int read_count = db_io_.gcount();
        if (read_count < chunk_size) {
            LOG_DEBUG("Read less than a page");
            db_io_.clear();
            memset((void *) (chunk_data.data() + read_count), 0, chunk_size - read_count);
        }
    }
}

int64_t DataChunkStorage::GetNextDataChunkOffset() {
    return file_size_;
}

void DataChunkStorage::ReadChunkData(int64_t chunk_offset, int64_t chunk_size, char *chunk_data) {
    // check if read beyond file length;
    if (chunk_offset > file_size_) {
        LOG_DEBUG("I/O error reading past end of file");
        LOG_DEBUG("datafile size: %ld, offset: %ld", file_size_, chunk_offset);
    } else {
        db_io_.seekp(chunk_offset);
        db_io_.read(chunk_data, chunk_size);
        if (db_io_.bad()) {
            LOG_DEBUG("I/O error while reading");
            return;
        }
        // if file ends before reading PAGE_SIZE
        int read_count = db_io_.gcount();
        if (read_count < chunk_size) {
            LOG_DEBUG("Read less than a page");
            db_io_.clear();
            memset(chunk_data + read_count, 0, chunk_size - read_count);
        }
    }
}

}